//
// Created by china on 25-1-17.
//

#include "Redis.h"

Redis::Redis(): _publishContext(nullptr),_subcribeContext(nullptr) {

}

Redis::~Redis() {
    if(_publishContext != nullptr){
        redisFree(_publishContext);
    }
    if(_subcribeContext != nullptr){
        redisFree(_subcribeContext);
    }
}

bool Redis::connect(){
    _publishContext = redisConnect("127.0.0.1",6379);
    if(_publishContext == nullptr){
        LOG_ERROR<< "_publishContext can not connect redis server";
        return false;
    }
    _subcribeContext = redisConnect("127.0.0.1",6379);
    if(_subcribeContext == nullptr){
        LOG_ERROR<< "_subcribeContext can not connect redis server";
        return false;
    }
    //在单独的线程中，监听通道上的事件，有消息给业务层进行上报
    std::thread t([&](){
        observerChannalMessage();
    });
    t.detach();
    LOG_INFO<<"connect redis server success";
    return true;
}

bool Redis::publish(int channel, std::string message) {
    LOG_INFO<<"Redis::publish";
    redisReply* reply = (redisReply*) redisCommand(_publishContext,"PUBLISH %d %s",channel,message.c_str());
    if(nullptr == reply){
        LOG_ERROR<<"fail to publish message";
        return false;
    }
    freeReplyObject(reply);
    return true;
}

bool Redis::subscribe(int channel) {
    LOG_INFO<<"Redis::subscribe";
    if(REDIS_ERR == redisAppendCommand(this->_subcribeContext,"SUBSCRIBE %d",channel)){
        LOG_ERROR<<"fail to subscribe";
        return false;
    }
    int done = 0;
    while(!done){
        if(REDIS_ERR == redisBufferWrite(this->_subcribeContext,&done)){
            LOG_ERROR<<"fail to subscribe";
            return false;
        }
    }
    return true;
}

bool Redis::unsubscibe(int channel) {
    LOG_INFO<<"Redis::unsubscibe";
    if(REDIS_ERR == redisAppendCommand(this->_subcribeContext,"UNSUBSCRIBE %d",channel)){
        LOG_ERROR<<"fail to unsubscibe";
        return false;
    }
    int done = 0;
    while(!done){
        if(REDIS_ERR == redisBufferWrite(this->_subcribeContext,&done)){
            LOG_ERROR<<"fail to unsubscibe";
            return false;
        }
    }
    return true;
}

void Redis::observerChannalMessage() {
    LOG_INFO<<"Redis::observerChannalMessage";
    redisReply* reply = nullptr;
    while(REDIS_OK == redisGetReply(this->_subcribeContext,(void**)&reply)){
        if(reply != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr){
            _notifyMessageHandler(atoi(reply->element[1]->str),reply->element[2]->str);
        }
        freeReplyObject(reply);
    }
    LOG_ERROR<<">>>>>>>>>>>>>>>>>>>> _notifyMessageHandler quit <<<<<<<<<<<<<<<<<<<<";
}

void Redis::initNotifyHandler(std::function<void(int, std::string)> fn) {
    this->_notifyMessageHandler = fn;
}
